package com.stf.rocketmq.controller;

import com.stf.rocketmq.entity.MsgTest;
import com.stf.rocketmq.util.MqUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;



/**
 * 同步消息发送
 * @author shangtf
 * */
@RestController
@RequestMapping("/syncMq")
@Slf4j
public class SyncProducerController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate; //注入Mq


    /**
     * ①发送普通同步消息-Object
     * ②发送普通同步消息-Object，并设置发送超时时间
     * */
    @RequestMapping("/sync/send1")
    public String syncSendString()  {

        //发送一个同步消息，会返回值 ---发送到 stringTopic 主题
        SendResult sendResult = rocketMQTemplate.syncSend(MqUtil.async_topic_string, "Hello RocketMQ");
        log.info("发送普通同步消息-Object，syncSendString===>{}", sendResult);

        //发送普通同步消息-Object，并设置发送超时时间
        //SendResult sendResult = rocketMQTemplate.syncSend("stringTopicWithTimeout", "Hello RocketMQ With timeOut" ,20000L);
        return sendResult.toString();
    }


    /**
     * 发送普通同步消息-Message,topic带tags
     * */
    @RequestMapping("/sync/send2")
    public String syncSendStringTags() {
        //syncSend和send是等价的
        rocketMQTemplate.syncSend(MqUtil.async_topic_string_tags, "hello world test2");
        //send底层还是会调用syncSend的代码
        rocketMQTemplate.send(MqUtil.async_topic_string_tags, MessageBuilder.withPayload("hello world test1").build());
        SendResult sendResult = rocketMQTemplate.syncSend(MqUtil.async_topic_string_tags+":"+"tags1", "hello world test2");
        log.info("发送普通同步消息-Message,topic带tags，syncSendStringTags===>{}", sendResult);
        return sendResult.toString();
    }


    /**
     * 发送普通同步消息-msg
     * 这里存在两种消息体，一种是Object的，另一种是Message<?>的形式的，其实我们发送Object的时候，底层是有帮我们做转换的
     * */
    @RequestMapping("/sync/send3")
    public String syncSendMessage()  {
        MsgTest msg = new MsgTest(1,"hello world test3",new Date());
        //发送一个同步消息，会返回值 ---发送到 stringTopic 主题
        SendResult sendResult = rocketMQTemplate.syncSend(MqUtil.async_topic_msg, MessageBuilder.withPayload(msg).build());
        log.info("发送普通同步消息-msg，syncSendMessage===>{}", sendResult);

        //发送普通同步消息-Message，并设置发送超时时间
        //SendResult sendResult = rocketMQTemplate.syncSend("first-topic-pojo", MessageBuilder.withPayload(msg).build(),20000L);
        return sendResult.toString();
    }



    /**
     * ①发送批量普通同步消息
     * ②发送批量普通同步消息，并设置发送超时时间
     * 使用Template发送批量消息时，发送的消息必须是org.springframework.messaging.Message的子类。批量发消息对消息的内容长度有限制，最大为4M，如果超过4M只能分为多批次发送。
     * */
    @RequestMapping("/sync/send4")
    public String syncSendCollectionMessage()  {

        // 方式一
        List<GenericMessage> list = new ArrayList<>();
        list.add(new GenericMessage("1"));
        list.add(new GenericMessage("2"));

        SendResult sendResult = rocketMQTemplate.syncSend(MqUtil.async_topic_collection, list);
        log.info("发送批量普通同步消息，syncSendCollectionMessage===>{}", sendResult);

        // 方式二：使用生产者批量发送，Message是org.apache.rocketmq.common.message.Message
        //rocketMQTemplate.getProducer().send(Collection<Message> msgs);

        //发送批量普通同步消息，并设置发送超时时间
        //SendResult sendResult = rocketMQTemplate.syncSend("collectionMessage-topic", list,20000L);
        return sendResult.toString();


       /* Message<String> msg = MessageBuilder.withPayload("hello world test1").build();
        List<Message> msgList = Arrays.asList(msg,msg,msg,msg,msg);
        SendResult res = rocketMQTemplate.syncSend("first-topic-str:tag1", msgList);
        log.info("批量消息");*/

    }


    //发送普通同步延迟消息，并设置超时，这个下文会演示

    /**
     * 同步延迟消息
     * rocketMQ的延迟消息发送其实是已发送就已经到broker端了，然后消费端会延迟收到消息。
     * RocketMQ 目前只支持固定精度的定时消息。
     * 固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * 延迟的底层方法是用定时任务实现的。
     */
    @RequestMapping("/sync/send5")
    public String syncSendDelayedStr()  {
        Message<String> message = MessageBuilder.withPayload("syncSendDelayedStr" + new Date()).build();
        /**
         * @param destination formats: `topicName:tags`
         * @param message 消息体
         * @param timeout 发送超时时间
         * @param delayLevel 延迟级别  1到18
         * @return {@link SendResult}
         */
        SendResult res = rocketMQTemplate.syncSend(MqUtil.async_delayed_str+":"+"tags2", message, 3000, 3);
        log.info("res==>{}", res);
        return res.toString();
    }


}
